package com.gmall.data.service;

import com.gmall.data.api.UserProfileQueryService;
import com.gmall.data.common.config.Config;
import com.gmall.data.common.utils.Constants;
import com.gmall.data.pojo.RuleParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Set;

/**
 * 用户画像查询服务，hbase查询实现类
 */
@Slf4j
public class UserProfileQueryServiceImpl implements UserProfileQueryService {

    Connection conn;
    Table table;

    public UserProfileQueryServiceImpl() {
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", Config.hbaseZookeeperQuorum());

            conn = ConnectionFactory.createConnection();
            table = conn.getTable(TableName.valueOf(Constants.USER_FEATURE_TABLE()));
        } catch (IOException e) {
            log.error("failed to createConnection hbase,{}", e);
        }
    }


    /**
     * 传入一个用户设备号，以及要查询的规则条件
     * 返回这些条件是否满足:
     * TODO 本查询只返回了成立与否，而查询到的画像数据值并没有返回 可能为将来的缓存模块带来不便，有待改造
     *
     * @param deviceId  设备id
     * @param ruleParam 规则参数对象
     * @return 是否成立
     */
    @Override
    public boolean judgeProfileCondition(String deviceId, RuleParam ruleParam) {

        // 1. 从规则参数中取出画像标签属性条件
        HashMap<String, String> userProfileParams = ruleParam.getUserProfileParams();

        // 2. 取出条件中所要求的待查询的标签名
        Set<String> tagNames = userProfileParams.keySet();

        // 3. 构造HBase的查询参数对象
        Get get = new Get(deviceId.getBytes());

        // 4. 把要查询的标签（hbase表中的列）逐一添加到get参数中
        for (String tagName : tagNames) {
            get.addColumn("f".getBytes(), tagName.getBytes());
        }

        // 5. 调用HBase Api执行查询
        try {
            Result result = table.get(get);

            // 对标签列表遍历
            for (String tagName : tagNames) {
                byte[] value = result.getValue("f".getBytes(), tagName.getBytes());
                // 如果有一列没有在HBase中查到，直接return false
                if (value == null)
                    return false;

                // 如果查询的结果和规则的值不匹配，也返回false
                String valueStr = new String(value);
                if (!valueStr.equals(userProfileParams.get(tagName)))
                    return false;
            }
            return true;
        } catch (Exception e) {
            log.error("failed to UserProfileQueryServiceImpl.judgeProfileCondition,devide={},userProfileParams={},message={}", deviceId, userProfileParams, e);
        }

        // 如果到了这，说明前面没有正常返回，也直接返回false
        return false;
    }
}
